package com.taotao.container;

import com.alibaba.fastjson.JSON;


import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.LinkedBlockingDeque;


@Component
public class LogContainer {
/**
 *
 * 去kibana查看数据使用下面的
 GET _search
 {
 "query": {
 "match": {
 "_index":"mayikt-log*"
 }
 }
 }
 */
    /**
     * 本地队列
     */
    private static LinkedBlockingDeque<String> logMsgs = new LinkedBlockingDeque<>();
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void log(String msg) {
        logMsgs.offer(msg);
    }

    public LogContainer() {
        new LogThread().start();
    }

    /**
     * 单独创建一个线程 从我们的队列中获取到 这些日志内容 在 异步的形式 投递到kafka中
     */
    class LogThread extends Thread {
        @Override
        public void run() {
            while (true) {
                String msgLog = logMsgs.poll();
                if (!StringUtils.isEmpty(msgLog)) {
                    // 投递kafka中
                    // kafkaTemplate.send("mayikt-log", msgLog);
                    kafkaTemplate.send("mayikt-log", msgLog).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable throwable) {
                            System.out.println("error{} get throwable msg:{}" + throwable.getMessage());
                        }

                        @Override
                        public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                            System.out.println("{}, success" + stringObjectSendResult.getProducerRecord().value());
                        }
                    });
                   /* kafkaTemplate.send("mayikt-log", JSON.toJSONString(msgLog)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                             @Override
                        public void onFailure(Throwable throwable) {
                            if (throwable instanceof KafkaProducerException) {
                                String value = (String) ((KafkaProducerException) throwable).getProducerRecord().value();
                                System.out.println("{} get throwable msg:{}"+value+throwable.getMessage());

                            } else {
                                System.out.println("get throwable msg:{}"+ throwable.getMessage());
                                //log.info("get throwable msg:{}", throwable.getMessage());
                            }
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> o) {
                            System.out.println("{}, success"+ o.getProducerRecord().value());
                            log.info("{}, success", o.getProducerRecord().value());
                        }
                    });*/


                }
                // 为了避免cpu飙高的问题
                try {
                    Thread.sleep(100);
                } catch (Exception e) {

                }
            }
        }
    }

}
